package club.janna.jrpc.register.redis;

import club.janna.jrpc.annotation.SPIService;
import club.janna.jrpc.register.Register;
import club.janna.jrpc.register.SubscribeCallback;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPubSub;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author Panbo.Guo
 */
@Slf4j
@SPIService("redis")
public class RedisRegister implements Register {

    /**
     * 服务提供者前缀
     */
    private final static String PREFIX_PROVIDER = "JRPC-provider-";

    /**
     * 服务消费者前缀
     */
    private final static String PREFIX_CONSUMER = "JRPC-consumer-";

    /**
     * 提供者前缀长度
     */
    private final static int LEN_PREFIX_PROVIDER = PREFIX_PROVIDER.length();

    /**
     * 搜索key的规则
     */
    private final static String KEYS_PATTERN = PREFIX_PROVIDER + "*%s*";

    /**
     * 订阅前缀
     */
    private final static String PREFIX_SUBSCRIBE = "__keyspace@0__:";

    /**
     * 订阅前缀的长度
     */
    private final static int LEN_PREFIX_SUBSCRIBE = PREFIX_SUBSCRIBE.length();

    /**
     * 订阅key模板
     */
    private final static String TEMPLATE_SUBSCRIBE = PREFIX_SUBSCRIBE + KEYS_PATTERN;

    /**
     * 是否已初始化
     */
    private volatile boolean init = false;

    /**
     * 超时时间，秒
     */
    private final static int TIMEOUT = 15;

    /**
     * 心跳间隔，秒
     */
    private final static int INTERVAL = 5;

    private JedisPool jedisPool;

    private Map<URI, String> serviceMap;

    private ScheduledExecutorService scheduledExecutorService;

    @Override
    public void init(URI register) {
        if(init)
            return;
        init = true;
        jedisPool = new JedisPool(register.getHost(), register.getPort());
        serviceMap = new HashMap<>();
        scheduledExecutorService = Executors.newScheduledThreadPool(10);
        scheduledExecutorService.scheduleAtFixedRate(this::heartbeat, INTERVAL, INTERVAL, TimeUnit.SECONDS);
    }

    @Override
    public void registerProvider(String name, URI uri) {
        register(name, uri, PREFIX_PROVIDER);
    }

    @Override
    public void registerConsumer(String name, URI uri) {
        register(name, uri, PREFIX_CONSUMER);
    }

    /**
     * 注册到redis
     * @param name
     * @param uri
     * @param prefix
     */
    private void register(String name, URI uri, String prefix) {
        serviceMap.put(uri, prefix);
        Jedis jedis = jedisPool.getResource();
        try {
            log.info("register to redis, uri -> {}", uri);
            jedis.setex(toKey(prefix, uri), TIMEOUT, prefix);
        } finally {
            jedis.close();
        }
    }

    @Override
    public Set<URI> fetch(String name) {
        Jedis jedis = jedisPool.getResource();
        try {
            Set<String> services = jedis.keys(String.format(KEYS_PATTERN, name));
            log.info("fetch service provider -> {}", services);
            if(services != null) {
                return services.stream().map(str -> {
                    try {
                        return new URI(str.substring(LEN_PREFIX_PROVIDER));
                    } catch (URISyntaxException e) {
                        log.warn("转换URI出错", e);
                    }
                    return null;
                }).filter(Objects::nonNull).collect(Collectors.toSet());
            }
        } finally {
            jedis.close();
        }
        return Collections.emptySet();
    }

    @Override
    public void subscribe(String name, SubscribeCallback callback) {
        String keysPattern = String.format(TEMPLATE_SUBSCRIBE, name);
        scheduledExecutorService.submit(() -> {
            log.info("开始订阅 -> {}", keysPattern);
            jedisPool.getResource().psubscribe(new JedisPubSub() {
                @Override
                public void onPMessage(String pattern, String channel, String message) {
                    log.debug("收到订阅消息, pattern -> {}, channel -> {},  message -> {}", pattern, channel, message);
                    URI uri = null;
                    try {
                        uri = new URI(channel.substring(LEN_PREFIX_SUBSCRIBE + LEN_PREFIX_PROVIDER));
                    } catch (URISyntaxException e) {
                        throw new RuntimeException(e);
                    }

                    if ("set".equals(message))
                        callback.callback(uri, SubscribeCallback.Action.UPDATE);
                    else if("expired".equals(message))
                        callback.callback(uri, SubscribeCallback.Action.REMOVE);
                }
            }, keysPattern);
        });
    }

    /**
     * 转换成redis key
     * @param uri
     * @return
     */
    private String toKey(String prefix, URI uri) {
        return prefix + uri.toString();
    }

    /**
     * 心跳
     */
    private void heartbeat() {
        if(CollectionUtils.isEmpty(serviceMap)) {
            log.warn("ready to send heartbeat, but no service!");
            return;
        }
        Jedis jedis = jedisPool.getResource();
        try {
            for (Map.Entry<URI, String> entry : serviceMap.entrySet()) {
                log.debug("renewal to redis, uri -> {}", entry.getKey());
                jedis.setex(toKey(entry.getValue(), entry.getKey()), TIMEOUT, entry.getValue());
            }
        } finally {
            jedis.close();
        }
    }
}
